/**
 * MCP Streamable HTTP 传输实现
 *
 * 实现MCP协议的Streamable HTTP传输层
 * 支持流式响应和Server-Sent Events (SSE)
 */

import { MCPProtocolHandler } from './mcp-protocol-handler.js';
import { logger, createErrorLogContext } from '../utils/logger.js';

export interface StreamableHttpOptions {
  port?: number;
  cors?: boolean;
  path?: string;
}

/**
 * Streamable HTTP 服务器
 */
export class MCPStreamableHttpServer {
  private handler: MCPProtocolHandler;
  private options: Required<StreamableHttpOptions>;

  constructor(handler: MCPProtocolHandler, options: StreamableHttpOptions = {}) {
    this.handler = handler;
    this.options = {
      port: options.port || 3001,
      cors: options.cors !== false,
      path: options.path || '/mcp',
    };
  }

  /**
   * 启动Streamable HTTP服务器
   */
  async start(): Promise<void> {
    const server = Bun.serve({
      port: this.options.port,
      fetch: async request => {
        const url = new URL(request.url);

        // CORS headers
        const headers: Record<string, string> = {
          'Content-Type': 'application/json',
        };

        if (this.options.cors) {
          headers['Access-Control-Allow-Origin'] = '*';
          headers['Access-Control-Allow-Methods'] = 'GET, POST, OPTIONS';
          headers['Access-Control-Allow-Headers'] = 'Content-Type, Accept';
        }

        // Handle preflight
        if (request.method === 'OPTIONS') {
          return new Response(null, { status: 204, headers });
        }

        // Streamable MCP endpoint
        if (url.pathname === this.options.path) {
          if (request.method === 'POST') {
            // 普通POST请求处理
            return await this.handlePost(request, headers);
          } else if (
            request.method === 'GET' &&
            request.headers.get('accept')?.includes('text/event-stream')
          ) {
            // SSE流式响应
            return await this.handleSSE(request);
          }
        }

        // Health check
        if (url.pathname === '/health') {
          return new Response(JSON.stringify({ status: 'ok', transport: 'streamable-http' }), {
            headers,
          });
        }

        return new Response('Not Found', { status: 404 });
      },
    });

    console.log(`MCP Streamable HTTP Server running at http://localhost:${this.options.port}`);
    console.log(`MCP endpoint: http://localhost:${this.options.port}${this.options.path}`);
    console.log(
      `SSE endpoint: GET http://localhost:${this.options.port}${this.options.path} (Accept: text/event-stream)`
    );
  }

  /**
   * 处理POST请求
   */
  private async handlePost(request: Request, headers: Record<string, string>): Promise<Response> {
    try {
      const body = await request.text();

      // 解析请求
      let requests: any[];
      try {
        const parsed = JSON.parse(body);
        requests = Array.isArray(parsed) ? parsed : [parsed];
      } catch (parseError) {
        logger.error('MCP Streamable HTTP: JSON parse error', createErrorLogContext(parseError, {
          body: body.substring(0, 1000) // 截取前1000字符避免日志过大
        }));
        return new Response(
          JSON.stringify({
            jsonrpc: '2.0',
            error: { code: -32700, message: 'Parse error' },
          }),
          { status: 400, headers }
        );
      }

      // 处理批量请求
      const responses = await Promise.all(requests.map(req => this.handler.handleRequest(req)));

      // 返回响应
      const responseBody = responses.length === 1 ? responses[0] : responses;
      return new Response(JSON.stringify(responseBody), { headers });
    } catch (error) {
      logger.error('MCP Streamable HTTP: Internal error in handlePost', createErrorLogContext(error, {
        method: 'handlePost'
      }));
      return new Response(
        JSON.stringify({
          jsonrpc: '2.0',
          error: {
            code: -32603,
            message: `Internal error: ${error instanceof Error ? error.message : String(error)}`,
          },
        }),
        { status: 500, headers }
      );
    }
  }

  /**
   * 处理SSE请求（Server-Sent Events）
   */
  private async handleSSE(request: Request): Promise<Response> {
    const stream = new ReadableStream({
      async start(controller) {
        // 发送初始化事件
        const encoder = new TextEncoder();
        controller.enqueue(encoder.encode(`data: ${JSON.stringify({ type: 'connected' })}\n\n`));

        // 保持连接活跃
        const keepAlive = setInterval(() => {
          try {
            controller.enqueue(encoder.encode(': keepalive\n\n'));
          } catch (keepAliveError) {
            logger.warn('MCP Streamable HTTP: keepalive error', createErrorLogContext(keepAliveError));
            clearInterval(keepAlive);
          }
        }, 30000);

        // 处理请求关闭
        request.signal.addEventListener('abort', () => {
          clearInterval(keepAlive);
          controller.close();
        });
      },
    });

    return new Response(stream, {
      headers: {
        'Content-Type': 'text/event-stream',
        'Cache-Control': 'no-cache',
        Connection: 'keep-alive',
        'Access-Control-Allow-Origin': '*',
      },
    });
  }
}

/**
 * 创建双向流式通信通道
 */
export class MCPStreamableChannel {
  private handler: MCPProtocolHandler;
  private encoder = new TextEncoder();
  private decoder = new TextDecoder();

  constructor(handler: MCPProtocolHandler) {
    this.handler = handler;
  }

  /**
   * 创建双向流
   */
  createBidirectionalStream(): {
    readable: ReadableStream;
    writable: WritableStream;
  } {
    let controller: ReadableStreamDefaultController;

    const readable = new ReadableStream({
      start(ctrl) {
        controller = ctrl;
      },
    });

    const writable = new WritableStream({
      write: async chunk => {
        try {
          // 解析输入
          const text = typeof chunk === 'string' ? chunk : this.decoder.decode(chunk);
          const request = JSON.parse(text);

          // 处理请求
          const response = await this.handler.handleRequest(request);

          // 发送响应
          controller.enqueue(this.encoder.encode(JSON.stringify(response) + '\n'));
        } catch (error) {
          logger.error('MCP Streamable HTTP: Error in bidirectional stream write', createErrorLogContext(error, {
            chunk: typeof chunk === 'string' ? chunk.substring(0, 500) : '[Binary data]'
          }));
          // 发送错误响应
          const errorResponse = {
            jsonrpc: '2.0',
            error: {
              code: -32700,
              message: 'Parse error',
            },
          };
          controller.enqueue(this.encoder.encode(JSON.stringify(errorResponse) + '\n'));
        }
      },

      close() {
        controller.close();
      },
    });

    return { readable, writable };
  }

  /**
   * 处理WebSocket连接（用于实时双向通信）
   */
  async handleWebSocket(ws: any): Promise<void> {
    ws.on('message', async (message: string | Buffer) => {
      try {
        const text = typeof message === 'string' ? message : message.toString();
        const request = JSON.parse(text);
        const response = await this.handler.handleRequest(request);
        ws.send(JSON.stringify(response));
      } catch (error) {
        logger.error('MCP Streamable HTTP: WebSocket message handling error', createErrorLogContext(error, {
          message: typeof message === 'string' ? message.substring(0, 500) : '[Binary message]'
        }));
        ws.send(
          JSON.stringify({
            jsonrpc: '2.0',
            error: {
              code: -32700,
              message: 'Parse error',
            },
          })
        );
      }
    });

    ws.on('close', () => {
      console.log('WebSocket connection closed');
    });
  }
}
